package com.yeming.flink.practice.transformation

import com.yeming.flink.practice.source.{MyCustomerSource, StationLog}
import org.apache.flink.streaming.api.scala._

object TranReduce {

  def main(args: Array[String]): Unit = {

    //环境设置
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    streamEnv.setParallelism(1)
    //添加数据源
    val stream: DataStream[StationLog] = streamEnv.addSource(new MyCustomerSource)
    //数据转换
    val result: DataStream[(String, Long)] = stream.filter(_.callType.equals("success"))
      .map(log => {
        (log.sid, log.duration)
      })
      .keyBy(0)
      .reduce((t1, t2) => {
        (t1._1, t1._2 + t2._2)
      })
    //数据sink
    result.print()
    //执行数据流
    streamEnv.execute("执行数据流转换")

  }

}
